package cn.itcast.streaming.async;

import cn.itcast.streaming.entity.ItcastDataPartObj;
import cn.itcast.streaming.entity.VehicleLocationModel;
import cn.itcast.streaming.utils.GaoDeMapUtils;
import cn.itcast.streaming.utils.GeoHashUtil;
import cn.itcast.streaming.utils.RedisUtil;
import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.async.ResultFuture;
import org.apache.flink.streaming.api.functions.async.RichAsyncFunction;
import org.apache.hadoop.hbase.util.Bytes;
import org.apache.http.HttpEntity;
import org.apache.http.HttpResponse;
import org.apache.http.client.config.RequestConfig;
import org.apache.http.client.methods.HttpGet;
import org.apache.http.impl.nio.client.CloseableHttpAsyncClient;
import org.apache.http.impl.nio.client.HttpAsyncClients;
import org.apache.http.util.EntityUtils;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.util.Collections;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.function.Supplier;

/**
 * Author itcast
 * Date 2021/9/27 15:09
 * Desc 通过异步请求获取指定经纬度的位置信息，从高德api获取位置数据
 * 将指定vin某个时间的位置数据保存到redis中
 */
public class AsyncHttpQueryFunction extends RichAsyncFunction<ItcastDataPartObj,ItcastDataPartObj> {
    private static final Logger logger = LoggerFactory.getLogger(AsyncHttpQueryFunction.class);

    //实现读取异步请求的客户端
    private static CloseableHttpAsyncClient httpAsyncClient = null;
    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        //配置HttpAsyncClient的配置
        RequestConfig config = RequestConfig.custom()
                .setConnectTimeout(5000)
                .setSocketTimeout(3000)
                .build();
        //初始化异步Http的client
        httpAsyncClient = HttpAsyncClients.custom()
                .setMaxConnTotal(5)
                .setDefaultRequestConfig(config)
                .build();
        //开启异步http的客户端
        httpAsyncClient.start();
    }
    //实现读取高德Api获取位置数据并将位置数据保存到redis中并返回ItcastDataPartObj
    @Override
    public void asyncInvoke(ItcastDataPartObj input, ResultFuture<ItcastDataPartObj> resultFuture) throws Exception {
        //4.1.获取当前车辆的经纬度
        Double lng = input.getLng();
        Double lat = input.getLat();
        //4.2.通过GaoDeMapUtils工具类根据参数获取请求的url
        String urlByLonLat = GaoDeMapUtils.getUrlByLonLat(lng, lat);
        //4.3.创建 http get请求对象
        HttpGet httpGet = new HttpGet(urlByLonLat);
        //4.4.使用刚创建的http异步客户端执行 http请求对象
        Future<HttpResponse> future = httpAsyncClient.execute(httpGet, null);
        //4.5.从执行完成的future中获取数据，返回ItcastDataPartObj对象
        CompletableFuture<ItcastDataPartObj> completableFuture = CompletableFuture.supplyAsync(new Supplier<ItcastDataPartObj>() {
            @Override
            public ItcastDataPartObj get() {
                //4.5.1.重写get方法
                try {
                    //定义需要返回的变量
                    String country = null;
                    String province = null;
                    String city = null;
                    String district = null;
                    String address = null;

                    HttpResponse httpResponse = future.get();
                    //4.5.1.1.使用future获取到返回的值
                    if (httpResponse.getStatusLine().getStatusCode() == 200) {
                        HttpEntity entity = httpResponse.getEntity();

                        //①通过IO流工具类直接生成字符串
                        /*String content1 = IOUtils.toString(entity.getContent());

                        //②通过将InputStream转换成输入Reader
                        InputStreamReader inputStreamReader = new InputStreamReader(entity.getContent());
                        //在读取数据流到buffer缓冲区
                        BufferedReader bufferedReader = new BufferedReader(inputStreamReader);
                        final int bufferSize = 1024;
                        final char[] buffer = new char[bufferSize];
                        final StringBuilder out = new StringBuilder();
                        for (; ; ) {
                            int rsz = bufferedReader.read(buffer, 0, buffer.length);
                            if (rsz < 0)
                                break;
                            out.append(buffer, 0, rsz);
                        }
                        String content2 = out.toString();*/
                        String jsonResult = EntityUtils.toString(entity);
                        //将json字符串转换成对象然后读取出来国家、省市区、address
                        JSONObject jsonObject = JSON.parseObject(jsonResult);
                        JSONObject regeocode = jsonObject.getJSONObject("regeocode");
                        if (regeocode != null && regeocode.size() > 0) {
                            address = regeocode.getString("formatted_address");
                            JSONObject addressComponent = regeocode.getJSONObject("addressComponent");
                            if (addressComponent != null && addressComponent.size() > 0) {
                                country = addressComponent.getString("country");
                                province = addressComponent.getString("province");
                                city = addressComponent.getString("city");
                                district = addressComponent.getString("district");
                                //将其封装为VehicleLocationModel 并写入到redis
                                VehicleLocationModel vehicleLocationModel = new VehicleLocationModel(
                                        country,
                                        province,
                                        city,
                                        district,
                                        address,
                                        lat,
                                        lng
                                );
                                //获取geohash值作为存储到redis的key
                                String geoHash = GeoHashUtil.encode(lat, lng);
                                RedisUtil.set(
                                        Bytes.toBytes(geoHash),
                                        vehicleLocationModel.toJsonStringArr()
                                );
                                //将当前车辆的位置信息赋值
                                input.setCountry(country);
                                input.setProvince(province);
                                input.setCity(city);
                                input.setDistrict(district);
                                input.setAddress(address);
                            } else {
                                logger.error("当前解析出来的地理信息为空，请检查!");
                            }
                        } else {
                            logger.error("当前解析出来的对象为空，请检查！");
                        }

                    } else {
                        logger.error("当前url请求返回response错误！");
                    }
                } catch (Exception e) {
                    e.printStackTrace();
                }
                return input;
            }
        });

        //4.6.从future的thenAccept
        completableFuture.thenAccept(new Consumer<ItcastDataPartObj>() {
            //4.6.1.重写accept方法，使用集合中只放一个对象
            @Override
            public void accept(ItcastDataPartObj itcastDataPartObj) {
                resultFuture.complete(Collections.singleton(itcastDataPartObj));
            }
        });
    }
    //如果当前请求超时，打印输出超时日志或告警信息
    @Override
    public void timeout(ItcastDataPartObj input, ResultFuture<ItcastDataPartObj> resultFuture) throws Exception {
        //超时时间，打印输出异步请求的超时警告
        System.out.println("当前异步请求超时！");
    }
    //关闭当前的http异步请求客户端
    @Override
    public void close() throws Exception {
        if(httpAsyncClient.isRunning()) httpAsyncClient.close();
    }
}
